package com.yuandengta.yomall.seckill.service.impl;

import com.alibaba.csp.sentinel.Entry;
import com.alibaba.csp.sentinel.SphU;
import com.alibaba.csp.sentinel.annotation.SentinelResource;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.fasterxml.jackson.core.type.TypeReference;
import com.yuandengta.common.to.mq.SeckillOrderTO;
import com.yuandengta.common.utils.JsonUtils;
import com.yuandengta.common.utils.R;
import com.yuandengta.common.vo.MemberVO;
import com.yuandengta.yomall.seckill.feign.CouponFeignService;
import com.yuandengta.yomall.seckill.feign.ProductFeignService;
import com.yuandengta.yomall.seckill.interceptor.LoginInterceptor;
import com.yuandengta.yomall.seckill.service.SeckillService;
import com.yuandengta.yomall.seckill.to.SeckillSkuRedisTO;
import com.yuandengta.yomall.seckill.vo.SeckillSessionsWithSkusVO;
import com.yuandengta.yomall.seckill.vo.SkuInfoVO;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RSemaphore;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * 秒杀服务 {@link SeckillServcieImpl}
 *
 * @author Kevin
 * @email: 178676392@qq.com
 */
@Slf4j
@Service
public class SeckillServcieImpl implements SeckillService {

    private final String SESSIONS_CACHE_PREFIX = "seckill:sessions:";
    private final String SECKILL_SKU_CACHE_PREFIX = "seckill:skus";
    private final String SKU_STOCK_SEMAPHORE = "seckill:stock:";

    private final String SECKILL_ORDER_EVENT_EXCHANGE = "order-event-exchange";
    private final String SECKILL_ORDER_QUEUE_ROUTING_KEY = "order.seckill.order";

    @Autowired
    private CouponFeignService couponFeignService;

    @Autowired
    private ProductFeignService productFeignService;

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 上架最近三天的秒杀商品
     */
    @Override
    public void uploadSeckillSkuLast3Days() {
        // 扫描最近三天需要参与的秒杀活动与商品信息
        R r = couponFeignService.getLast3DaySession();
        if (r.getCode() == 0) {
            // 需要上架的商品
            List<SeckillSessionsWithSkusVO> sessions = r.getData("data",
                    new TypeReference<List<SeckillSessionsWithSkusVO>>() {
            });
            // 缓存秒杀活动信息
            saveSessions(sessions);
            // 缓存秒杀活动关联的商品信息
            saveSessionSkus(sessions);
        } else {
            log.error("远程调用 yomall-coupon 获取秒杀活动失败");
        }
    }

    private List<SeckillSkuRedisTO> getCurrentSeckillSkusBlockHandler(BlockException e) {
        log.error("getCurrentSeckillSkusResource被限流了");
        return null;
    }

    /**
     * 获取当前时间可以参与的秒杀商品信息
     * @return
     */
    @SentinelResource(value = "getCurrentSeckillSkus", blockHandler = "getCurrentSeckillSkusBlockHandler")
    @Override
    public List<SeckillSkuRedisTO> getCurrentSeckillSkus() {
        // 确定当前时间属于哪个秒杀场次
        long currentTime = System.currentTimeMillis();
        Set<String> keys = redisTemplate.keys(SESSIONS_CACHE_PREFIX + "*");
        for (String key : keys) {
            String[] split = key.replace(SESSIONS_CACHE_PREFIX, "").split("_");
            long startTime = Long.parseLong(split[0]);
            long endTime = Long.parseLong(split[1]);
            if (currentTime >= startTime && currentTime <= endTime) {
                // 获取这个秒杀场次需要的所有商品信息
                List<String> range = redisTemplate.opsForList().range(key, -100, 100);
                BoundHashOperations<String, String, String> hashOps = redisTemplate.boundHashOps(SECKILL_SKU_CACHE_PREFIX);
                List<String> list = hashOps.multiGet(range);
                if (list != null && list.size() > 0) {
                    List<SeckillSkuRedisTO> redisTOS = list.stream().map(item -> {
                        SeckillSkuRedisTO redisTO = JsonUtils.jsonToPojo(item, SeckillSkuRedisTO.class);
                        return redisTO;
                    }).collect(Collectors.toList());
                    return redisTOS;
                }
                break;
            }
        }

        return null;
    }

    /**
     * 获取商品秒杀信息
     * @param skuId
     * @return
     */
    @Override
    public SeckillSkuRedisTO getSkuSeckillInfo(Long skuId) {
        // 找到所有需要参与秒杀商品的key
        BoundHashOperations<String, String, String> hashOps = redisTemplate.boundHashOps(SECKILL_SKU_CACHE_PREFIX);
        Set<String> keys = hashOps.keys();
        if (keys != null && keys.size() > 0) {
            String regx = "\\d_" + skuId;
            for (String key : keys) {
                if (Pattern.matches(regx, key)) {
                    String json = hashOps.get(key);
                    SeckillSkuRedisTO redisTO = JsonUtils.jsonToPojo(json, SeckillSkuRedisTO.class);
                    // 处理随机码，只要在秒杀时间段内才返回随机码
                    long currentTime = System.currentTimeMillis();
                    if (!(currentTime >= redisTO.getStartTime() && currentTime <= redisTO.getEndTime())) {
                        redisTO.setRandomCode(null);
                    }

                    return redisTO;
                }
            }
        }

        return null;
    }

    /**
     * 秒杀
     * @param killId 秒杀场次id_商品id
     * @param key 随机码
     * @param num 商品数量
     * @return
     */
    @Override
    public String seckill(String killId, String key, Integer num) {
        MemberVO memberVO = LoginInterceptor.threadLocal.get();

        // 获取当前秒杀商品信息
        BoundHashOperations<String, String, String> hashOps = redisTemplate.boundHashOps(SECKILL_SKU_CACHE_PREFIX);
        String json = hashOps.get(killId);
        if (StringUtils.isEmpty(json)) {
            return null;
        }

        SeckillSkuRedisTO redisTO = JsonUtils.jsonToPojo(json, SeckillSkuRedisTO.class);

        // 校验合法性
        // 1.校验时间
        long currentTime = System.currentTimeMillis();
        if (currentTime >= redisTO.getStartTime() && currentTime <= redisTO.getEndTime()) {
            // 2.校验随机码和商品id
            String randomCode = redisTO.getRandomCode();
            String skuId = redisTO.getPromotionSessionId() + "_" + redisTO.getSkuId();
            if (randomCode.equals(key) && skuId.equals(killId)) {
                // 3.校验购物数量
                if (num <= redisTO.getSeckillLimit().intValue()) {
                    // 4.验证是否购买过
                    // 幂等性，只要秒杀成功，就去redis占位 SETNX，userId_sessionId_skuId
                    String redisKey = memberVO.getId() + "_" + skuId;
                    Long ttl = redisTO.getEndTime() - redisTO.getStartTime();
                    Boolean ifAbsent = redisTemplate.opsForValue()
                            .setIfAbsent(redisKey, num.toString(), ttl, TimeUnit.MILLISECONDS);
                    if (ifAbsent) {
                        // 占位成功，说明没有购买过
                        RSemaphore semaphore = redissonClient.getSemaphore(SKU_STOCK_SEMAPHORE + randomCode);
                        // 快速尝试
                        boolean acquire = semaphore.tryAcquire(num);
                        if (acquire) {
                            // 秒杀成功，快速下单，发消息到MQ
                            String orderSn = IdWorker.getTimeId();
                            SeckillOrderTO seckillOrderTO = new SeckillOrderTO();
                            seckillOrderTO.setOrderSn(orderSn);
                            seckillOrderTO.setMemberId(memberVO.getId());
                            seckillOrderTO.setNum(num);
                            seckillOrderTO.setPromotionSessionId(redisTO.getPromotionSessionId());
                            seckillOrderTO.setSkuId(redisTO.getSkuId());
                            seckillOrderTO.setSeckillPrice(redisTO.getSeckillPrice());

                            rabbitTemplate.convertAndSend(SECKILL_ORDER_EVENT_EXCHANGE,
                                    SECKILL_ORDER_QUEUE_ROUTING_KEY, seckillOrderTO);

                            return  orderSn;
                        }
                    }
                }
            }
        }

        return null;
    }

    /**
     * 缓存秒杀活动信息
     * @param sessions
     */
    private void saveSessions(List<SeckillSessionsWithSkusVO> sessions) {
        if (sessions != null && sessions.size() > 0) {
            sessions.stream().forEach(session -> {
                Long startTime = session.getStartTime().getTime();
                long endTime = session.getEndTime().getTime();
                String key = SESSIONS_CACHE_PREFIX + startTime + "_" + endTime;
                Boolean hasKey = redisTemplate.hasKey(key);
                if (!hasKey) {
                    // 缓存秒杀活动信息
                    List<String> skuIds = session.getRelationSkus().stream()
                            .map(item -> item.getPromotionSessionId() + "_" + item.getSkuId())
                            .collect(Collectors.toList());
                    redisTemplate.opsForList().leftPushAll(key, skuIds);
                }
            });
        }
    }

    /**
     * 缓存秒杀活动关联的商品信息
     * @param sessions
     */
    private void saveSessionSkus(List<SeckillSessionsWithSkusVO> sessions){
        if (sessions != null && sessions.size() > 0) {
            sessions.stream().forEach(session -> {
                BoundHashOperations<String, Object, Object> hashOps = redisTemplate.boundHashOps(SECKILL_SKU_CACHE_PREFIX);
                session.getRelationSkus().stream().forEach(seckillSkuVO -> {
                    String hashKey = seckillSkuVO.getPromotionSessionId() + "_" + seckillSkuVO.getSkuId();
                    if (!hashOps.hasKey(hashKey)) {
                        // 缓存商品
                        SeckillSkuRedisTO redisTO = new SeckillSkuRedisTO();
                        // 1.sku基本信息
                        R r = productFeignService.getSkuInfo(seckillSkuVO.getSkuId());
                        if (r.getCode() == 0) {
                            SkuInfoVO skuInfo = r.getData("skuInfo", new TypeReference<SkuInfoVO>() {
                            });
                            redisTO.setSkuInfo(skuInfo);
                        }

                        // 2.sku秒杀信息
                        BeanUtils.copyProperties(seckillSkuVO, redisTO);

                        // 3.设置当前商品秒杀时间
                        redisTO.setStartTime(session.getStartTime().getTime());
                        redisTO.setEndTime(session.getEndTime().getTime());

                        // 4.设置随机码，防止恶意攻击
                        String token = UUID.randomUUID().toString().replace("-", "");
                        redisTO.setRandomCode(token);

                        String json = JsonUtils.objectToJson(redisTO);
                        hashOps.put(hashKey, json);

                        // 5.使用秒杀商品库存作为分布式的信号量，限流
                        RSemaphore semaphore = redissonClient.getSemaphore(SKU_STOCK_SEMAPHORE + token);
                        semaphore.trySetPermits(seckillSkuVO.getSeckillCount().intValue());
                    }
                });
            });
        }
    }
}
